package com.ww.springboot.boot.algorithm.滑动窗口;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import rx.Observable;
import rx.Observer;
import rx.Subscription;
import rx.functions.Func1;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * rxJava 滑动窗口示例
 */
@Slf4j
public class WindowDemo {

    /**
     * 演示模拟Hystrix的健康统计metric
     */
    @Test
    public void hystrixTimewindowDemo() throws InterruptedException {
        //创建Random类对象
        Random random = new Random();

        //模拟Hystrix event事件流，每100毫秒发送一个0或1随机值
        //随机值为0代表失败，随机值为1代表成功
        Observable<Long> observable = Observable
                //间隔
                .interval(100, TimeUnit.MILLISECONDS)
                //造数据
                .map(i -> random.nextInt(2))
                //窗口大小
                .window(3000, TimeUnit.MILLISECONDS)
                //统计窗口数据 对每个窗口内的数据进行加工
                .flatMap((Func1<Observable<Integer>, Observable<Long>>) eventBucket -> {
//                    Observable<List<Integer>> olist = eventBucket.toList();
//                    olist.forEach(list ->
//                    {
//                        long count = list.stream().filter(i -> i == 0).count();
//                        log.info("{}-{} '0 count:{}", Thread.currentThread(), list.toString(), count);
//                    });
//                    Observable<Long> skip = eventBucket.scan((o1, o2) -> o1 + o2).map(o->Long.valueOf(o)).skip(29);
//                    return skip;
                    Observable<List<Integer>> olist = eventBucket.toList();
                    Observable<Long> countValue = olist.map(list ->
                    {
                        long count = list.stream().filter(i -> i == 0).count();
                        log.info("{}-{} '0 count:{}", Thread.currentThread(), list.toString(), count);
                        return count;

                    });
                    return countValue;
                })
//                //滑动窗口区间大小
//                .window(3, 3)
//                //统计滑动窗口数据 对指定的多个窗口汇总数据进行加工
//                .flatMap((Func1<Observable<Long>, Observable<Long>>) eventBucket -> eventBucket.reduce((bucket1, bucket2) -> {
//                    /**
//                     *对窗口内的桶进行累加
//                     */
//                    return bucket1 + bucket2;
//                }))
                //被订阅时 调用一次 (当开启share()时 只会被调用一次 )
                .doOnSubscribe(() -> log.info("------------------doOnSubscribe--------------------"))
                //取消订阅时 调用一次(当开启share()时 只有所有订阅者都取消才会调此方法)
                .doOnUnsubscribe(() -> log.info("------------------doOnUnsubscribe--------------------"))
                //订阅者共享数据
                .share();

        //observable.subscribe(sum -> log.info("{} 滑动窗口的和1：{}", Thread.currentThread(), sum));
        //observable.subscribe(sum -> log.info("{} 滑动窗口的和2：{}", Thread.currentThread(), sum));
        //observable.subscribe(sum -> log.info("滑动窗口的和3：{}", sum));
        //observable.subscribe(sum -> log.info("滑动窗口的和4：{}", sum));
        //observable.subscribe(sum -> log.info("滑动窗口的和5：{}", sum));

        Observer<Long> observer = new Observer() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Object o) {
                log.info("{} 滑动窗口的和1：{}", Thread.currentThread(), o);
            }
        };
        Observer<Long> observer2 = new Observer() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Object o) {
                log.info("{} 滑动窗口的和2：{}", Thread.currentThread(), o);
            }
        };

        Subscription subscribe = observable.subscribe(observer);
        Subscription subscribe1 = observable.subscribe(observer2);

        System.out.println("------------------------------");

        Thread.sleep(10000);
        subscribe.unsubscribe();
        //subscribe1.unsubscribe();

        Thread.sleep(Integer.MAX_VALUE);
    }
}
